package handle

import (
	"github.com/nsqio/go-nsq"
)

type nsqHandleMessage = func(msg *nsq.Message, err error)

type NsqHandle[T any] struct {
	info    string
	data    T
	handles []nsqHandleMessage
}

func (m *NsqHandle[T]) HandleMessage(msg *nsq.Message) error {
	var err error
	for _, item := range m.handles {
		item(msg, err)
	}
	return err
}
func (m *NsqHandle[T]) SetData(data T) *NsqHandle[T] {
	m.data = data
	return m
}
func (m *NsqHandle[T]) SetMsg(info string) *NsqHandle[T] {
	m.info = info
	return m
}

func NewNsqHandle[T any](data T, handles ...nsqHandleMessage) *NsqHandle[T] {
	return &NsqHandle[T]{data: data, handles: handles}
}

type NsqConsumer struct {
	addrs    []string
	consumer *nsq.Consumer
}

func NewConsumer(address []string) (c *NsqConsumer) {
	return &NsqConsumer{addrs: address}
}

func (n *NsqConsumer) InitConsumer(topic string, channel string, config *nsq.Config) *NsqConsumer {
	var err error
	n.consumer, err = nsq.NewConsumer(topic, channel, config)
	if err != nil {
		panic(err)
	}
	return n
}

func (n *NsqConsumer) AddHandler(handler ...nsq.Handler) {
	for _, item := range handler {
		n.consumer.AddHandler(item)
	}
}

// 直接连NSQD
func (n *NsqConsumer) ConnectToNSQDs() error {
	if err := n.consumer.ConnectToNSQDs(n.addrs); err != nil {
		//if err := c.ConnectToNSQLookupd(address); err != nil { // 通过lookupd查询
		panic(err)
	}
	return nil
}

// 通过lookupd查询
func (n *NsqConsumer) ConnectToNSQLookupds() error {
	if err := n.consumer.ConnectToNSQLookupds(n.addrs); err != nil {
		//if err := c.ConnectToNSQLookupd(address); err != nil { // 通过lookupd查询
		panic(err)
	}
	return nil
}

func (n *NsqConsumer) AddConcurrentHandlers(handler nsq.Handler, concurrency int) {
	n.consumer.AddConcurrentHandlers(handler, concurrency)
}
